package com.example.mq.mqserver.core;

import com.example.mq.VirtualHost;
import com.example.mq.common.Consumer;
import com.example.mq.common.ConsumerEnv;
import com.example.mq.common.MqException;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;


public class ConsumerManager {
    private VirtualHost parent;

    //指定一个线程池，执行具体任务
    private ExecutorService workerPool = Executors.newFixedThreadPool(4);

    //存放令牌的阻塞队列
    private BlockingQueue<String> tokenQueue = new LinkedBlockingQueue<>();

    //扫描线程，用来获取发送了的消息
    private Thread scannerThread;

    public ConsumerManager(VirtualHost parent) {
        this.parent = parent;
         scannerThread =new Thread(()->{
             // 这块应该是一个后台线程，不应该是前台线程，前台线程，这里肯定走不下去了。这个应该是辅助线程，当你另一个代码携带了令牌走到这里，然后触发欧克
             while (true) {
                 try {
                     //1、拿到令牌(阻塞队列)
                     String token = tokenQueue.take();//无线阻塞下去
//                String token=tokenQueue.poll();
                     System.out.println("拿到了令牌！！！！！！！！！queueName="+token);

                     //2、根据令牌，找到队列
                     MSGQueue queue = parent.getMemoryDataCenter().getQueue(token);

                     if (queue == null) {
                         throw new MqException("根据令牌获取队列，但是没有找到队列 tooken(队列名字):" + token);
                     }
                     synchronized (queue) {
                         consumerMessage(queue);
                     }
                 } catch (Exception e) {
                     throw new RuntimeException(e);
                 }
             }
         });
         scannerThread.setDaemon(true);//设置为后台线程
        scannerThread.start();//启动线程（辅助线程）
    }

    //队列名字放到了阻塞队列，线程将会苏醒
    public void notifyConsumer(String queueName) throws InterruptedException {
        tokenQueue.put(queueName);
    }


    //添加一个消费者
    public void addConsumer(String consumerTag, String queueName, boolean autoAck, Consumer consumer) throws MqException {
        MSGQueue queue = parent.getMemoryDataCenter().getQueue(queueName);
        if (queue == null) {
            throw new MqException("[ConsumerManager] 队列不存在! queueName=" + queueName);
        }
        ConsumerEnv consumerEnv = new ConsumerEnv(consumerTag, queueName, autoAck, consumer);
        synchronized (queue) {
            queue.addConsumer(consumerEnv);

            //如果队列中有消息(可能是已有的消费者处理不过来，或者之前没有这个队列订阅的消费者，产生了消息积压，所以这里需要把消息全部都消费掉)
            int n = parent.getMemoryDataCenter().getMessageCount(queueName);
            for (int i = 0; i < n; i++) {
                consumerMessage(queue);
            }
        }
    }

    //用于当前队列从订阅消费者中选择一个消费者，消费消息
    private void consumerMessage(MSGQueue queue) throws MqException {

        //前台线程不结束，进程无法结束
        //后台线程不会影响进程的结束

        //1、按照轮询的方式获取消费者
        ConsumerEnv luckDog = queue.getConsumerEnv();//轮询方式获取一个队列中的消息
        if (luckDog == null) {
            //如果当前队列没有消费者，那就不消费，直接返回
            System.out.println("当前队列没有消费者，那就不消费，直接返回");
            return;
        }

        //2、根据队列名字，获取该队列中的一个消息
        Message message = parent.getMemoryDataCenter().pollMessage(queue.getName());

        //3、把这个消息，给到线程池子，执行回调delivery
        workerPool.submit(() -> {
            try {
                //1、消息放入待确认集合
                parent.getMemoryDataCenter().addMessageWaitAck(queue.getName(),message);

                //2、执行回调 赋值handleDelivery参数
                luckDog.getConsumer()
                        .handleDelivery
                                (luckDog.getConsumerTag()
                                        , message.getBasicProperties()
                                        , message.getBody());
                System.out.println("[ConsumerManager] 消息被成功消费! queueName=" + queue.getName());

                //3、自动应答，直接删除消息
                if(luckDog.isAutoAck()){
                    //1）删除硬盘上的消息
                    //二是，消息持久化
                    if(message.getDeliverMode()==2){
                        parent.getDiskDataCenter().deleteMessage(queue,message);
                    }

                    //2）删除内存中的消息
                    parent.getMemoryDataCenter().deleteMessage(message.getMessageId());


                    //3)删除待确认消息
                    parent.getMemoryDataCenter().removeMessageWaitAck(queue.getName(), message  );
                }

                //4、不是自动应答，需要bsicACK手动确认
            } catch (Exception e) {
                e.printStackTrace();
            }

        });

    }
}
